package main

import (
	"bufio"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/url"
	"os"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/influxdata/influxdb/client/v2"
)

type Reader interface {
	Read(rc chan []byte)
}

type Writer interface {
	Write(wc chan *Message)
}

type logProcess struct {
	rc    chan []byte
	wc    chan *Message
	read  Reader
	write Writer
}

type ReadFromFile struct {
	path string //读取文件的路径
}

func (rff *ReadFromFile) Read(rc chan []byte) {
	//读取模块
	f, err := os.OpenFile(rff.path, os.O_CREATE|os.O_RDWR, 0755)
	if err != nil {
		panic(fmt.Sprintf("open file error: %s", err.Error()))
	}
	// 从文件末尾逐行读取文件内容
	f.Seek(0, 2)

	rd := bufio.NewReader(f)
	for {
		line, err := rd.ReadBytes('\n')
		if err == io.EOF {
			time.Sleep(500 * time.Millisecond)
			continue
		} else if err != nil {
			panic(fmt.Sprintf("readbytes error :%s", err.Error()))
		}
		// 用来监控 读取的 日志条数 每读取一条，往这个channel中添加一个标记。
		TypeMonitor <- TypeHandleLen
		rc <- line[:len(line)-1]
	}

}

type WriteToInfluxDB struct {
	influxDBDsn string // influx data source
}

func (wti *WriteToInfluxDB) Write(wc chan *Message) {
	infSlic := strings.Split(wti.influxDBDsn, "@")
	// 创建一个httpclient
	c, err := client.NewHTTPClient(client.HTTPConfig{
		Addr:     infSlic[0],
		Username: infSlic[1],
		Password: infSlic[2],
	})
	if err != nil {
		log.Fatal(err)
	}
	defer c.Close()

	for v := range wc {
		// Create a new point batch
		bp, err := client.NewBatchPoints(client.BatchPointsConfig{
			Database:  infSlic[3],
			Precision: infSlic[4], //写入精度 这里是秒
		})
		if err != nil {
			log.Fatal(err)
		}
		//Fields: UpstreamTime, RequestTime, BytesSent
		//Time : TimeLocal
		// Tags: Path, Method, Scheme, Status
		// Create a point and add to batch
		tags := map[string]string{"Path": v.Path, "Method": v.Method, "Scheme": v.Scheme, "Status": v.Status}
		fields := map[string]interface{}{
			"UpstreamTime": v.UpstreamTime,
			"RequestTime":  v.RequestTime,
			"BytesSent":    v.BytesSent,
		}

		pt, err := client.NewPoint("nginx_log", tags, fields, v.TimeLocal)
		if err != nil {
			log.Fatal(err)
		}
		bp.AddPoint(pt)

		// Write the batch
		if err := c.Write(bp); err != nil {
			log.Fatal(err)
		}

		// Close client resources
		if err := c.Close(); err != nil {
			log.Fatal(err)
		}
		log.Println("write success")
	}
}

//以下的方法都是指针的原因：
// 1. 不用对结构体进行拷贝，当结构体不大的时候，优势不明显，一旦结构体足够大的时候，就会发现，用指针的比用值的快很多
// 2. 指针是可以修改自身的一些参数
type Message struct {
	TimeLocal                    time.Time
	BytesSent                    int
	Path, Method, Scheme, Status string
	UpstreamTime, RequestTime    float64
}

// 系统状态监控
const (
	TypeHandleLen = 0
	TypeErrNum    = 1
)

var TypeMonitor = make(chan int, 200)

type SystemInfo struct {
	HandleLine   int     `json:"handleline"`
	Tps          float64 `json:"tps"`
	ReadChanLen  int     `json:"readchanlen"`
	WriteChanLen int     `json:"writechanlen"`
	RunTime      string  `json:"runtime"`
	ErrNum       int     `json:"errnum"`
}

type Monitor struct {
	startTime time.Time
	data      SystemInfo
	tpsSli    []int
}

func (m *Monitor) start(lp *logProcess) {

	// 程序运行监控代码
	go func() {
		for n := range TypeMonitor {
			switch n {
			case TypeHandleLen:
				m.data.HandleLine += 1
			case TypeErrNum:
				m.data.ErrNum += 1
			}
		}
	}()

	// 吞吐量
	ticker := time.NewTicker(time.Second * 5) //这是一个五秒的时间间隔
	go func() {
		for {
			<-ticker.C //每五秒钟会触发一次，才能继续运行下去
			// 保证tpsSli中的数据只有前后五秒的。
			m.tpsSli = append(m.tpsSli, m.data.HandleLine)
			if len(m.tpsSli) > 2 {
				m.tpsSli = m.tpsSli[1:]
			}
		}
	}()

	http.HandleFunc("/monitor", func(rw http.ResponseWriter, r *http.Request) {
		m.data.RunTime = time.Now().Sub(m.startTime).String()
		m.data.ReadChanLen = len(lp.rc)
		m.data.WriteChanLen = len(lp.wc)
		if len(m.tpsSli) >= 2 {

			m.data.Tps = float64(m.tpsSli[1]) - float64(m.tpsSli[0])/5
		}

		ret, _ := json.MarshalIndent(m.data, "", "\t")

		io.WriteString(rw, string(ret))
		http.ListenAndServe(":9193", nil)
	})
}

//Influxdb--

func (lp *logProcess) Process() {
	//解析模块
	// 从Read Channel中读取每行日志数据

	//正则提取所需的监控数据（path，status，method等）
	// 172.0.0.12 - - [04/Mar/2018:13:48:53 +0000] http "GET /foo?query=t HTTP/1.0" 200 2133 "-" " KeepAAliveClient" "-" 1.005 1.854
	//([\d\.]+)\s+([^ \[]+)\s+([^ \[]+)\s+\[([^\]]+)\]\s+([a-z]+)\s+\"([^"]+)\"\s+(\d{3})\s+(\d+)\s+\"([^"]+)\"\s+\"(.*?)\"\s+\"([\d\.-]+)\"\s+([\d\.-]+)\s+([\d\.-]+)
	//r := regexp.MustCompile(`([\d\\.]+)\s+([^ \[]+)\s+([^ \[]+)\s+\[([^\\]]+)\\]\s+([a-z]+)\s+\\"([^"]+)\\"\s+(\d{3})\s+(\d+)\s+\\"([^"]+)\\"\s+\\"(.*?)\\"\s+\\"([\d\\.-]+)\\"\s+([\d\\.-]+)\s+([\d\\.-]+)`)
	// 开始接受管道中传来的byte切片
	//echo 输入字符到文件中，特殊字符需要进行转意哦，就是” 要用\"来代替
	r := regexp.MustCompile(`([\d\.]+)\s+([^ \[]+)\s+([^ \[]+)\s+\[([^\]]+)\]\s+([a-z]+)\s+\"([^"]+)\"\s+(\d{3})\s+(\d+)\s+\"([^"]+)\"\s+\"(.*?)\"\s+\"([\d\.-]+)\"\s+([\d\.-]+)\s+([\d\.-]+)`)
	loc, _ := time.LoadLocation("Asia/Shanghai")
	for v := range lp.rc {
		//根据正则表达式开始匹配，如果过好中的内容匹配到了，就以切片的形式返回
		ret := r.FindStringSubmatch(string(v))
		if len(ret) != 14 {
			TypeMonitor <- TypeErrNum
			log.Println("findstringsubmathc failed:", string(v))
			continue
		}
		message := &Message{}
		t, err := time.ParseInLocation("02/Jan/2006:15:04:05 +0000", ret[4], loc)
		if err != nil {
			TypeMonitor <- TypeErrNum
			log.Println("parseinlocation failed:", err.Error(), ret[4])
			continue
		}
		message.TimeLocal = t

		byteSent, _ := strconv.Atoi(ret[8])
		message.BytesSent = byteSent

		reqSli := strings.Split(ret[6], " ")
		if len(reqSli) != 3 {
			TypeMonitor <- TypeErrNum
			log.Println("strings.split failed:", ret[6])
			continue
		}
		message.Method = reqSli[0]

		u, err := url.Parse(ret[1])
		if err != nil {
			TypeMonitor <- TypeErrNum
			log.Println("url parse failed:", err.Error())
			continue
		}
		message.Path = u.Path

		message.Scheme = ret[5]

		message.Status = ret[7]

		upstreamtime, _ := strconv.ParseFloat(ret[12], 64)
		requestTime, _ := strconv.ParseFloat(ret[13], 64)
		message.RequestTime = requestTime
		message.UpstreamTime = upstreamtime

		lp.wc <- message

		// 写入Write Channel
	}
}

func main() {
	// 这四行代码是将命令行中的参数传递进来，用的事flag包
	var path, influxDsn string
	flag.StringVar(&path, "path", "./access.log", "read file path")
	flag.StringVar(&influxDsn, "influxDsn", "http://127.0.0.1:8086@gcc@123456@imooc@s", "influxdn conncet infomations")
	flag.Parse()

	r := &ReadFromFile{
		path: path,
	}
	w := &WriteToInfluxDB{
		influxDBDsn: influxDsn,
	}
	// 将channel 变成有缓存的channel
	lp := &logProcess{
		rc:    make(chan []byte, 200),
		wc:    make(chan *Message, 200),
		read:  r,
		write: w,
	}
	// 因为读日志最快，处理中等，写入最慢，所以为了达到平衡，可以多个gorutine执行一个任务
	// go lp.read.Read(lp.rc)
	// go lp.Process()
	// go lp.write.Write(lp.wc)
	go lp.read.Read(lp.rc)
	for i := 0; i < 2; i++ {
		go lp.Process()
	}
	for i := 0; i < 4; i++ {
		go lp.write.Write(lp.wc)
	}

	// time.Sleep(300000 * time.Second)
	m := Monitor{
		startTime: time.Now(),
		data:      SystemInfo{},
	}
	m.start(lp)
}
